﻿
#include "threadpool.h"
#include <functional>
#include <thread>
#include <iostream>
#include <thread>
#include <iostream>

const int TASK_MAX_THREAD_HOLD = INT32_MAX;
const int TASK_MAX_THRESHHOLD = 100;
const int THREAD_MAX_IDLE_TIME = 60; // 单位：秒

/* 
项目中除了0/1不能出现其他的魔鬼数字，用有意义的名字来代替
*/
ThreadPool::ThreadPool()
	: initThreadSize_(0)
	, idleThreadsize_(0)
	, curThreadsize_(0)
	, threadSizeThreshHold_(TASK_MAX_THRESHHOLD)
	, taskQueueMaxThreadHold_(TASK_MAX_THREAD_HOLD)
	, taskSize_(0)
	, poolMode_(PoolMode::MODE_FIXED)
	, isPoolRunning_(false)
{}

/*
析构函数，回收线程资源；
*/
ThreadPool::~ThreadPool()
{
	isPoolRunning_ = false;
	// 等待线程池里面所有的线程返回  有两种状态：阻塞 & 正在执行任务中
	// 析构是用户线程使用的，线程函数是线程池里的线程，不同线程之间的通信
	std::unique_lock<std::mutex> lock(taskQueMtx_);
	notEmpty_.notify_all();   // 唤醒等待的线程，把等待的线程在线程函数中释放掉
	//编译运行的时候发现无法退出，因为这里exitCond_没有被唤醒，阻塞了
	exitCond_.wait(lock, [&]()->bool {return threads_.size() == 0; });
}

/* 
设置线程池的工作模式
*/
void ThreadPool::setMode(PoolMode mode)
{
	if (checkRunningState())  // 不允许启动之后进行修改
		return;

	poolMode_ = mode;
}

/*
设置task任务队列上线阈值，让用户自己设置这个阈值
*/
void ThreadPool::setTaskQueMaxThreshHold(int threadhold)
{
	if (checkRunningState())
		return;

	taskQueueMaxThreadHold_ = threadhold;
}

void ThreadPool::setThreadSizeThreshHold(int threshhold)
{
	if (checkRunningState())
		return;

	if (poolMode_ == PoolMode::MODE_CACHED)
	{
		threadSizeThreshHold_ = threshhold;
	}
}

/*
给线程池提交任务 接口设计成智能指针封装的任务出来
用户调用该接口，传入任务对象，生产任务
*/
Result ThreadPool::submitTask(std::shared_ptr<Task> sp)
{
	// 获取锁 用户提交任务操作任务队里，线程消费任务也操作任务队列，做线程互斥
	// unique_lock的构造函数里会将传入的这把锁调用lock();  _Pmtx->lock();
	std::unique_lock<std::mutex> lock(taskQueMtx_);

	// 线程通信 等待（任务队列有空余）能产生任务，放到队列里
	// 如果任务的个数等于任务队列中的阈值，不能继续提交了，
	// 细节处理：用户提交任务，阻塞1s时间，任务队列还没有空闲下来，那就返回任务提交失败（服务降级，给用户合理的提示）
	// notFull_处理方式一：
	/*
	while (taskQueueMaxThreadHold_ == taskQue_.size())
	{
		// notFull_情况下才能放任务，现这里任务队列已经满了，所以需要在notFull_条件上进行wait
		// 当前线程进入等待状态
		notFull_.wait(lock);
	}
	*/
	// notFull_处理方式二：
	// 提示的第二个参数：predicate 是一个函数对象，用lamda表达式表示即可
	// 在lamda表达式中要访问外部成员变量，捕获值用&即可,满足条件，就可继续往下执行，否则处于等待状态
	//notFull_.wait(lock, [&]()->bool {return taskQue_.size() < taskQueueMaxThreadHold_; });
	// notFull_处理方式三：
	// 线程的通信  等待任务队列有空余 c++11:  wait   wait_for   wait_until
	// wait: 一直等，等到lamda表达式中条件满足的时候再起来，继续向下执行
	// wait_for: lamda条件满足,立刻返回。1s时间到了，条件满足/条件不满足，均返回，区分这两种方式，根据wait_for的返回值判断 
	// wait_until: 加了时间参数，我等你到下周一，这个等待是设置的时间终止节点，
	// 用户提交任务，最长不能阻塞超过1s，否则判断提交任务失败，返回
	if (!notFull_.wait_for(lock, std::chrono::seconds(1), 
		[&]()->bool { return taskQue_.size() < (size_t)taskQueueMaxThreadHold_;} ))
	{
		// 表示notFull_等待1s种，条件依然没有满足
		std::cerr << "task queue is full, submit task fail." << std::endl;
		//return;   //返回Result对象
		// return task->getResult();  // Task  Result  返回task的成员方法，线程执行完task，task对象就被析构掉了，
									  // 之后用户还要继续使用这个对象
		return Result(sp, false);
	}
		
	// 如果任务队列有空余，把任务放到任务队列中
	taskQue_.emplace(sp);
	taskSize_++;
	// 因为新放了任务，任务肯定不空了，notEmpty_通知
	notEmpty_.notify_all();

	// cached模式问题：任务处理比较紧急 场景：小而快的任务 需要根据任务数量和空闲线程的数量，判断是否需要创建新的线程出来；
	// 耗时的任务多的，不适合cached模式，会长时间占用一个线程，如果使用cached模式，会导致创建线程过多而影响性能；
	if (poolMode_ == PoolMode::MODE_CACHED
		&& taskSize_ > (std::atomic_uint)idleThreadsize_
		&& curThreadsize_ < threadSizeThreshHold_)
	{
		std::cout << ">>> create new thread..." << std::endl;
		// 创建新的线程对象
		// threadFunc中增加了参数，因为要回收线程，这里的bind需要曾加一个占位符
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		int threadId = ptr->getId(); //根据ID识别线程，便于回收线程
		threads_.emplace(threadId, std::move(ptr));
		//创建完线程对象之后，要启动线程
		threads_[threadId]->start(); 
		//threads_.emplace_back(std::move(ptr));
		// 修改线程个数相关变量
		curThreadsize_++;
		idleThreadsize_++;
	}

	//return; //返回Result对象
	//return task->getResult();
	return Result(sp); // 默认true
}

/*  
开启线程池，直接获取CPU的核心个数；
*/
void ThreadPool::start(int initThreadSize)
{
	// 设置线程池的运行状态
	isPoolRunning_ = true;

	// 记录初始线程个数，默认为CPU核心数量，用户修改后使用用户修改的数据；
	initThreadSize_ = initThreadSize;
	curThreadsize_ = initThreadSize;
	
	// 创建线程对象，为了保证线程启动的公平特性，先集中创建线程对象，后启动所有线程
	// 这样线程再start时，就不需要创建对象了，绑定器和函数对象
	// thread对象start时候要执行threadpool里的函数，thread对象拿到threadFunc函数，才可以执行这个函数
	for (int i = 0; i < initThreadSize_; i++)
	{
		// bind(&ThreadPool::threadFunc，this) : 把ThreadPool对象指针绑定到threadFunc方法上，函数对象
		// Thread(std::bind() : ThreadPool中的threadFunc方法 绑定到Thread对象
		// 成员方法threadFunc绑定到了线程对象Thread，需要在线程对象的构造函数中接收

		// 优化
		// new出来的要释放，定义中使用指针接收，需要释放资源，就是用智能指针，不用手动释放资源
		//threads_.emplace_back(new Thread(std::bind(&ThreadPool::threadFunc, this)));
		// threadFunc中增加了参数，因为要回收线程，这里的bind需要曾加一个占位符,调用threadFunc需要主动传一个参数
		auto ptr = std::make_unique<Thread>(std::bind(&ThreadPool::threadFunc, this, std::placeholders::_1));
		// ptr放到vector中，形参到实参需要拷贝一份，调用了在源代码中显式标记为 = deleted 的成员函数
		// unique_ptr相对于auto_ptr删除了左值拷贝构造和赋值重载函数，增加了右值引用的拷贝构造和赋值，一个资源只能让一个智能指针来指向
		// 资源放到容器里，做资源转移move
		
		int threadId = ptr->getId(); //根据ID识别线程，便于回收线程
		threads_.emplace(threadId, std::move(ptr));
		//threads_.emplace_back(std::move(ptr));
	}
	
	// 启动所有线程 std::vector<Thread*> threads_;
	for (int i = 0; i < initThreadSize_; i++)
	{						   
		// 线程函数监视任务队列，有任务就抢，执行任务
		// 需要去执行一个线程函数，线程池中的每个线程要执行线程函数threadFunc
		threads_[i]->start();   //map提供了[]运算符重载函数，key作为下标，得到value值，使用没问题

		// 记录初始空闲线程的数量
		idleThreadsize_++; 
	}
}

/* 
线程函数：线程池的所有线程从任务队列里面消费任务
int threadid ：传入线程的id，在线程回收中使用，submitTask中绑定的时候需要增加参数占位符
*/
void ThreadPool::threadFunc(int threadid) // 线程函数返回，相应的线程也就结束了
{
	auto lastTime = std::chrono::high_resolution_clock().now();
	// 线程要循环从任务队列里取任务
	// 所有任务必须执行完成，在这里把任务处理完，线程池才可以回收所有线程资源
	for (;;)
	{
		std::shared_ptr<Task> task; // 默认构造为空
		{
			// 获取任务队列的锁 
			// unique_lock的构造函数里会将传入的这把锁调用lock();  _Pmtx->lock();
			std::unique_lock<std::mutex> lock(taskQueMtx_);

			// cached问题：
			// cached模式下，有可能已经创建了很多的线程，但是空闲时间超过60s，应该把多余的线程
			// 结束回收掉（超过initThreadSize_数量的线程要进行回收）
			// 当前时间 - 上一次线程执行的时间 > 60s
# if 0 
			if (poolMode_ == PoolMode::MODE_CACHED)
			{
				// 每一秒中返回一次   怎么区分：超时返回？还是有任务待执行返回?   点开wait_for看里面的实现
				// 判断超时返回,判断当任务队列没有任务时候
				while (taskQue_.size() == 0)
				{
					if (std::cv_status::timeout ==
						notEmpty_.wait_for(lock, std::chrono::seconds(1)))
					{
						//now的返回值time_point
						auto now = std::chrono::high_resolution_clock().now();
						//C++11封装比较灵活，可以在时间之间任意切换时间单位,这里需要得到秒单位的差值
						auto cur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
						if (cur.count() > THREAD_MAX_IDLE_TIME
							&& curThreadsize_ > initThreadSize_)
						{
							// 开始回收当前线程
							// 记录线程数量的相关变量的值修改
							// 把线程对象从线程列表容器中删除，无法匹配线程函数对应的线程对象 threadFunc《=》thread对象
							// 写东西边写边设计会造成问题，写之前要对方法进行详细的设计
							// 分析：threadid => thread对象 => 删除
							// 通过线程的id，那么每个线程就需要一个id
							// std::this_thread::get_id() 这是库生成的线程id； 这里是自己生成的id,注意传参
							threads_.erase(threadid);
							curThreadsize_--;
							idleThreadsize_--;

							std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
								<< std::endl;
							return;
						}
					}
				}
			}
			else
			{
				//等待notEmpty_条件，若任务队列为空，就一直等待
				notEmpty_.wait(lock, [&]()->bool { return taskQue_.size() > 0; });
			}
#endif

			// 每一秒中返回一次   怎么区分：超时返回？还是有任务待执行返回?   点开wait_for看里面的实现
			// 判断超时返回,判断当任务队列没有任务时候，任务执行完成之后
			// 锁 + 双重判断isPoolRunning_  处理死锁问题
			while (taskQue_.size() == 0)
			{
				if (!isPoolRunning_)
				{

					// 线程结束的时候，线程池的线程正在执行任务，再次回来的时候，进不去while循环；
					threads_.erase(threadid);
					std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
					<< std::endl;
					exitCond_.notify_all();
					return; //线程函数结束，线程结束
				}
					
				if (poolMode_ == PoolMode::MODE_CACHED)
				{
					if (std::cv_status::timeout ==
						notEmpty_.wait_for(lock, std::chrono::seconds(1)))
					{
						//now的返回值time_point
						auto now = std::chrono::high_resolution_clock().now();
						//C++11封装比较灵活，可以在时间之间任意切换时间单位,这里需要得到秒单位的差值
						auto cur = std::chrono::duration_cast<std::chrono::seconds>(now - lastTime);
						if (cur.count() > THREAD_MAX_IDLE_TIME
							&& curThreadsize_ > initThreadSize_)
						{
							// 开始回收当前线程
							// 记录线程数量的相关变量的值修改
							// 把线程对象从线程列表容器中删除，无法匹配线程函数对应的线程对象 threadFunc《=》thread对象
							// 写东西边写边设计会造成问题，写之前要对方法进行详细的设计
							// 分析：threadid => thread对象 => 删除
							// 通过线程的id，那么每个线程就需要一个id
							// std::this_thread::get_id() 这是库生成的线程id； 这里是自己生成的id,注意传参
							threads_.erase(threadid);
							curThreadsize_--;
							idleThreadsize_--;

							std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
								<< std::endl;
							return;
						}
					}
				}
				else
				{				
					//MODE_FIXED模式 等待notEmpty_条件，若任务队列为空，就一直等待
					notEmpty_.wait(lock);

				}
				//判断是有任务被唤醒还是线程池结束了要被唤醒
				// 线程池要结束，回收线程资源
				//if (!isPoolRunning_)
				//{
				//	threads_.erase(threadid);

				//	std::cout << "threadid:" << std::this_thread::get_id() << " exit!"
				//		<< std::endl;
				//	exitCond_.notify_all();
				//	return;
				//}
			}

			//std::cout << "tid : " << std::this_thread::get_id() << "尝试获取任务..." << std::endl;
			idleThreadsize_--; // 有任务可以取了，空闲线程的数量--

			std::cout << "tid : " << std::this_thread::get_id() << "获取任务成功..." << std::endl;

			// 从任务队列种取一个任务出来
			task = taskQue_.front();
			taskQue_.pop();
			taskSize_--;

			// 如果依然有剩余任务，notEmpty_继续通知其它的线程执行任务
			if (taskQue_.size() > 0)
			{
				notEmpty_.notify_all();
			}

			// 取出一个任务，notFull_通知可以继续提交生产任务
			notFull_.notify_all();
		}
		// 应该是拿到任务之后就把锁释放掉，而不是应该把任务处理完成之后再释放锁；
		// 锁的作用：保证操作任务队列的时候，任务队列里有任务；保证任务队列的线程安全操作；
		// 释放锁，别的任务也能去执行 / 用户也能提交任务； 利用局部对象出作用域就自动析构的特点

		// 当前线程负责执行这个任务，taskQue_中智能指针指向的都是task基类，需要调用从task继承而来的派生类方法
		// 基类指针指向那个对象，就会调用那个对象的同名覆盖方法；
		if (task != nullptr)
		{
			//task->run();//执行任务，把任务的返回值setVal方法给到Result,封装一个方法来处理即可
			task->exec();
		}
		
		idleThreadsize_++; // 线程的任务完成，空闲线程数量++
		// 如果依然有剩余任务，继续通知其它得线程执行任务
		// 取出一个任务，进行通知，通知可以继续提交生产任务

		lastTime = std::chrono::high_resolution_clock().now(); //更新线程执行完任务的时间
	}
}

// 更多的判断放在这个检查函数中来实现即可
bool ThreadPool::checkRunningState() const
{
	return isPoolRunning_;
}

/*
////////////////task方法的实现
*/
Task::Task()
	: result_(nullptr)
{}

void Task::exec()
{
	if (result_ != nullptr)
	{
		// 这里发生多态调用；把任务的返回值通过setVal方法给到Result
		result_->setVal(run()); 
	}
}

void Task::setResult(Result* res)
{
	result_ = res;
}

/*
////////////////线程方法的实现
*/
int Thread::generateId_ = 0;

Thread::Thread(ThreadFunc func)
	: func_(func)  // 接收绑定器绑定的线程函数threadFunc
	, threadId_(generateId_++)
{}

/*
析构函数
*/
Thread::~Thread()
{
}

/*
启动线程
*/
void Thread::start()
{
	// 创建一个线程来执行一个线程函数 pthread_create
	// C++11来说 线程对象t  和线程函数func_
	// 绑定的有参数，把threadid传递进去即
	std::thread t(func_, threadId_);

	// 设置分离线程（分离线程对象和线程函数执行过程），出作用域之后，线程对象析构，但不能让线程函数结束
	// 线程要虎视眈眈的等待去消费任务，整个应用程序不结束，线程池也不能结束
	// 主程序结束后，分离线程自动回收，
	t.detach(); 
}

int Thread::getId() const
{
	return threadId_;
}

/*
/////////////// Result的实现
*/
Result::Result(std::shared_ptr<Task> task, bool isValid)
	: task_(task)
	, isValid_(isValid)
{
	task_->setResult(this); // 把Result对象设置给task对象
}

// 问题二：get方法，用户调用这个方法获取task的返回值　 用户调用的
Any Result::get()
{
	if (!isValid_)
	{
		return "";
	}
	sem_.wait();// task任务如果没有执行完，这里会阻塞用户的线程
	return std::move(any_);
}

// 问题一：setVal方法，获取任务执行完的返回值的,task执行完之后，记录到any_中　谁调用的呢？？
void Result::setVal(Any any)
{
	this->any_ = std::move(any);
	// 已经获取的任务的返回值，增加信号量资源
	// Linux下的死锁：用户使用Result出作用域，对象就已析构了，信号量的相关资源已释放，调用post会造成死锁
	sem_.post(); 
}
